package com.cbj.netty.client;



import com.cbj.netty.server.MyCat;
import com.cbj.netty.util.RPCUtil;

import com.cbj.netty.handler.ClientResponseHandler;
import com.cbj.netty.handler.ServerDecoder;
import com.cbj.netty.protocol.Content;
import com.cbj.netty.protocol.Header;
import com.cbj.netty.rpc.ResponseCallBack;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.io.*;
import java.net.*;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 连接工厂负责创建对应的客户端连接
 */
public class ClientFactory{
    private ClientFactory(){
    }
    // 单例工厂
     public static final ClientFactory factory;
     static {
         factory = new ClientFactory();
     }
     public static ClientFactory getFactory(){
         return factory;
     }
     // 指定连接池的大小
    int poolSize = 10;
     // 用于从连接池中随机获取连接
    Random rand = new Random();

    // 首先一个客户端可以连接多个服务端，每一个服务端都有自己的pool，所以使用一个map集合,key是服务端地址，pool就是服务端的连接池
    // 用于获取要连接的服务端的连接池
    ConcurrentHashMap<InetSocketAddress, ClientPool> serverPool = new ConcurrentHashMap<>();


    /**
     * 核心方法:基于自定义协议发送请求体
     * @param content 请求体
     * @param type 使用的协议
     * @return 根据CompletableFuture的get方法拿到返回结果
     */
    public static CompletableFuture<Object> transport(Content content,String type){
        CompletableFuture<Object> res = new CompletableFuture<>();
        if(type.equals("rpc")){
            // 1.将contest和head转换为底层传输需要的字节数组
            byte[] msgBody = RPCUtil.getBytes(content);
            Header header = Header.createHeader(msgBody);
            byte[] msgHeader = RPCUtil.getBytes(header);
            // 2.通过连接池区取得连接
            NioSocketChannel clientChannel = factory.getClient(new InetSocketAddress("localhost", 9090));
            // 3.由于是异步调用，我们要将请求结果返回到requestId对应的线程
            long requestId = header.getRequestId();
            // 4.将requestId对应的线程放入回调池，等待结果的返回
            ResponseCallBack.addCallBack(requestId, res);
            // 5.开始发送数据
            ByteBuf byteBuf = RPCUtil.comHeadAndContent(msgHeader, msgBody);
            ChannelFuture channelFuture = clientChannel.writeAndFlush(byteBuf);

        }else {
            // 使用http协议
            urlTS(content,res);
        }
        return res;
    }


    /**
     * 根据服务器address从连接池中获取连接
     * @param address
     * @return
     */
    public NioSocketChannel getClient(InetSocketAddress address){
        ClientPool clientPool = serverPool.get(address);
        // client还没有初始化，双重检查
        if(clientPool == null){
            synchronized (serverPool){
                if(clientPool == null){
                    serverPool.putIfAbsent(address,new ClientPool(poolSize));
                    clientPool = serverPool.get(address);
                }
            }
        }
        int nextInt = rand.nextInt(poolSize);
        // 如果初始化过了，并且是活跃的，就直接取走
        if(clientPool.clients[nextInt] !=null && clientPool.clients[nextInt].isActive() ){
            return clientPool.clients[nextInt];
        }else {
            // 没有初始化，就将对应下标赋值，同时为了确保同时只有一个线程赋值，使用sync锁
            synchronized (clientPool.lock[nextInt]){
                if(clientPool.clients[nextInt] ==null || !clientPool.clients[nextInt].isActive())
                return clientPool.clients[nextInt] = create(address);
            }
        }
        return clientPool.clients[nextInt];
    }

    /**
     * 真正地创建客户端的逻辑
     * @param address
     * @return
     */
    private NioSocketChannel create(InetSocketAddress address){
        // 基于Netty的客户端
        NioEventLoopGroup clientWorker = new NioEventLoopGroup(1);
        ChannelFuture connect = new Bootstrap()
                .group(clientWorker)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new ServerDecoder()); // 解码器
                        pipeline.addLast(new ClientResponseHandler()); // 此handler中具体处理请求，返回结果
                    }
                }).connect(address);
        try {
            // 阻塞拿到客户端
            NioSocketChannel client = (NioSocketChannel) connect.sync().channel();
            return client;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 基于Http协议发送请求体
     * @param content
     * @param res
     */
    private static void urlTS(Content content, CompletableFuture<Object> res) {
        try {
            URL url = new URL("http://localhost:9090/");
            HttpURLConnection urlConnection = (HttpURLConnection)url.openConnection();
            // post请求
            urlConnection.setRequestMethod("POST");
            urlConnection.setDoOutput(true);
            urlConnection.setDoInput(true);

            OutputStream out = urlConnection.getOutputStream();
            ObjectOutputStream oout = new ObjectOutputStream(out);
            oout.writeObject(content);// 这里并没有真正发送
            Object serverRes = null;
            // 阻塞等待返回结果
            if(urlConnection.getResponseCode()==200){
                InputStream in = urlConnection.getInputStream();
                ObjectInputStream oin = new ObjectInputStream(in);
                Content o = (Content)oin.readObject();
                serverRes = o.getRes();
            }
            res.complete(serverRes);
        } catch (MalformedURLException e) {
            e.printStackTrace();
        }catch (IOException e) {
            e.printStackTrace();
        }catch (ClassNotFoundException e){
            e.printStackTrace();
        }

    }


}